package com.corn.flink.lesson3;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2023/2/24 11:24
 */

public class FlinkRichFunctionDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.fromElements(
                new Person("alice", 15, "cq"),
                new Person("jack", 18, "cq"),
                new Person("rose", 20, "sh"),
                new Person("mick", 42, "gz"),
                new Person("white", 12, "cq"),
                new Person("maria", 32, "sz"),
                new Person("candy", 56, "bj"),
                new Person("jerry", 28, "wh")
        ).map(new RichMapFunction<Person, Person>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                System.out.println("map open " + getRuntimeContext().getJobId() + " " + getRuntimeContext().getTaskName() + " task started!");
            }

            @Override
            public Person map(Person person) throws Exception {
                person.setAge(person.getAge() * 2);
                return person;
            }

            @Override
            public void close() throws Exception {
                super.close();
                System.out.println("map close " + getRuntimeContext().getJobId() + " " + getRuntimeContext().getTaskName() + "task closed!");
            }
        }).print();
        env.execute();
    }
}
